package cn.edu.flink.scala.tutorial.sql

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.bridge.scala._

object SelectTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource)

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    tEnv.createTemporaryView("click_table", clickStream)

    val table1 = tEnv.sqlQuery(
      s"""
         |select user, url
         |from click_table
         |""".stripMargin)


    tEnv.toDataStream(table1).print("ChangelogStream")

    env.execute("SelectTest")
  }
}
